package org.apache.hadoop.hdfs.server.datanode;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.security.PrivilegedExceptionAction;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.DFSConfigKeys;
import org.apache.hadoop.hdfs.DFSUtil;
import org.apache.hadoop.security.UserGroupInformation;

import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;

/**
 * 为DataNode节点管理BPOfferService对象。
 * 对于BPOfferService对象的创建、移除、启动、停止等操作必须通过该类的api来完成。
 *
 * BlockPoolManager主要维护的就是该DataNode上的BPOfferService对象，及其所属nameserviceId、blockPoolId。
 *
 * BlockPoolManager中主要包含如下几个数据结构：
 * 1、保存nameserviceId与BPOfferService的对应关系的HashMap：bpByNameserviceId；
 * 2、保存blockPoolId与BPOfferService的对应关系的HashMap：bpByBlockPoolId；
 * 3、保存所有BPOfferService的ArrayList：offerServices；
 * 4、DataNode实例dn；
 * 5、refreshNamenodes()方法中用于线程间同步或互斥锁的Object：refreshNamenodesLock。
 */
@InterfaceAudience.Private
class BlockPoolManager {
  private static final Log LOG = DataNode.LOG;

  /** NameserviceId与BPOfferService的对应关系 */
  private final Map<String, BPOfferService> bpByNameserviceId = Maps.newHashMap();
  /** BlockPoolId与BPOfferService的对应关系 */
  private final Map<String, BPOfferService> bpByBlockPoolId = Maps.newHashMap();
  /** 所有的BPOfferService */
  private final List<BPOfferService> offerServices = Lists.newArrayList();

  // DataNode实例dn
  private final DataNode dn;

  // 这个refreshNamenodesLock仅仅在refreshNamenodes()方法中被用作互斥锁
  private final Object refreshNamenodesLock = new Object();

  // 构造函数
  BlockPoolManager(DataNode dn) {
    this.dn = dn;
  }
  
  synchronized void addBlockPool(BPOfferService bpos) {
    Preconditions.checkArgument(offerServices.contains(bpos), "Unknown BPOS: %s", bpos);
    if (bpos.getBlockPoolId() == null) {
      throw new IllegalArgumentException("Null blockpool id");
    }
    bpByBlockPoolId.put(bpos.getBlockPoolId(), bpos);
  }
  
  /**
   * Returns the array of BPOfferService objects. 
   * Caution: The BPOfferService returned could be shutdown any time.
   */
  synchronized BPOfferService[] getAllNamenodeThreads() {
    BPOfferService[] bposArray = new BPOfferService[offerServices.size()];
    return offerServices.toArray(bposArray);
  }
      
  synchronized BPOfferService get(String bpid) {
    return bpByBlockPoolId.get(bpid);
  }
  
  synchronized void remove(BPOfferService t) {
    offerServices.remove(t);
    if (t.hasBlockPoolId()) {
      // It's possible that the block pool never successfully registered
      // with any NN, so it was never added it to this map
      bpByBlockPoolId.remove(t.getBlockPoolId());
    }
    
    boolean removed = false;
    for (Iterator<BPOfferService> it = bpByNameserviceId.values().iterator();
         it.hasNext() && !removed;) {
      BPOfferService bpos = it.next();
      if (bpos == t) {
        it.remove();
        LOG.info("Removed " + bpos);
        removed = true;
      }
    }
    
    if (!removed) {
      LOG.warn("Couldn't remove BPOS " + t + " from bpByNameserviceId map");
    }
  }
  
  void shutDownAll(BPOfferService[] bposArray) throws InterruptedException {
    if (bposArray != null) {
      for (BPOfferService bpos : bposArray) {
        bpos.stop(); //interrupts the threads
      }
      //now join
      for (BPOfferService bpos : bposArray) {
        bpos.join();
      }
    }
  }
  
  synchronized void startAll() throws IOException {
    try {
      // 遍历offerServices，启动所有的BPOfferService
      // 这块东西隐藏了极为关键的逻辑，这里包含了最核心datanode启动的时候，要向namenode去进行注册，让namenode感知到自己
      // 开源的hadoop源码，也不一定就是特别好的，datanode注册的这个事情，明明是最关键的，
      // 它们一定要隐藏BlockPoolManager组件里，这个入口的代码，还看起来很不起眼
      UserGroupInformation.getLoginUser().doAs(
          new PrivilegedExceptionAction<Object>() {
            @Override
            public Object run() throws Exception {
              // 这里启动的意思，就是BPServiceActor的线程
              // BPOfferService --> 一组Namendoe --> 2个BPServiceActor --> 负责跟1个namenode通信
              // 一旦start启动了两个后台的线程，BPServiceaActor对应了一个后台运行的线程
              // 这个后台运行的线程就是负责跟Namenode进行注册、心跳、汇报block,通信以及一些交互
              for (BPOfferService bpos : offerServices) {
                bpos.start(); // TODO
              }
              return null;
            }
          });
    } catch (InterruptedException ex) {
      IOException ioe = new IOException();
      ioe.initCause(ex.getCause());
      throw ioe;
    }
  }
  
  void joinAll() {
    for (BPOfferService bpos: this.getAllNamenodeThreads()) {
      bpos.join();
    }
  }
  
  void refreshNamenodes(Configuration conf) throws IOException {
    LOG.info("Refresh request received for nameservices: " + conf.get(DFSConfigKeys.DFS_NAMESERVICES));

    // 1. 从配置信息conf中获取nameserviceid->{namenode名称->InetSocketAddress}的映射集合newAddressMap
    Map<String, Map<String, InetSocketAddress>> newAddressMap = DFSUtil.getNNServiceRpcAddressesForCluster(conf); //

    // 需要通过使用synchronized关键字在refreshNamenodesLock上加互斥锁
    synchronized (refreshNamenodesLock) {
      // 2. 调用doRefreshNamenodes()方法执行集合newAddressMap中的刷新
      doRefreshNamenodes(newAddressMap);
    }
  }

  /**
   * 将需要处理的nameservice分别加入到不同的集合，然后按照添加、删除、更新的顺序针对处理类型相同的nameservice一并处理即可。
   *
   * 1、第一步，针对nameserviceid->{namenode名称->InetSocketAddress}的映射集合newAddressMap中每个nameserviceid，
   *       确认它是一个完全新加的nameservice，还是一个其NameNode列表被更新的nameservice，分别加入 待添加toAdd 和 待刷新toRefresh集合；
   *       它的处理思路是，
   *           循环addrMap中每个nameserviceid，放入待添加toAdd或者待刷新toRefresh集合；
   *           如果bpByNameserviceId结合中存在nameserviceId，加入待刷新集合toRefresh，否则加入到待添加集合toAdd。
   * 2、第二步，针对newAddressMap中没有，而目前DataNode内存bpByNameserviceId中存在的nameservice，需要删除，添加到待删除toRemove集合；
   *       它的处理思路是：
   *           利用Sets的difference()方法，比较bpByNameserviceId和addrMap两个集合的keySet，找出bpByNameserviceId中存在，但是addrMap中不存在的nameserviceid，生成待删除集合toRemove。
   * 3、第三步，处理 待添加toAdd集合，启动所有新的nameservices：根据addrs创建BPOfferService，维护BPOfferService相关映射集合，然后启动所有的BPOfferService；
   *       这一步针对待添加集合toAdd中的每个nameserviceId，做以下处理：
   *           3.1、从addrMap中根据nameserviceId获取对应Socket地址InetSocketAddress，创建集合addrs；
   *           3.2、根据addrs创建BPOfferService实例bpos；
   *           3.3、将nameserviceId->BPOfferService的对应关系添加到集合bpByNameserviceId中
   *           3.4、将BPOfferService添加到集合offerServices中；
   *        最后，调用startAll()方法启动所有BPOfferService，实际上是通过调用它的start()方法启动。
   * 4、第四步，处理 待删除toRemove集合，停止所有旧的nameservices；
   *        遍历待删除集合toRemove中的每个nameserviceId：
   *           4.1、根据nameserviceId从集合bpByNameserviceId中获取BPOfferService；
   *           4.2、调用BPOfferService的stop()和join()方法停止服务，它们会调用本身的remove()方法；
   * 5、第五步，处理 待刷新toRefresh集合，更新NN列表已变化的nameservices。
   *       遍历待更新集合toRefresh中的每个nameserviceId：
   *           5.1、根据nameserviceId从集合bpByNameserviceId中取出对应的BPOfferService；
   *           5.2、根据BPOfferService从配置信息addrMap中取出NN的Socket地址InetSocketAddress，形成列表addrs；
   *           5.3、调用BPOfferService的refreshNNList()方法根据addrs刷新NN列表。
   */
  private void doRefreshNamenodes(Map<String, Map<String, InetSocketAddress>> addrMap) throws IOException {
    // 确保当前线程在refreshNamenodesLock上拥有互斥锁
    assert Thread.holdsLock(refreshNamenodesLock);

    // 定义三个集合，分别为待刷新的toRefresh、待添加的toAdd和待移除的toRemove
    Set<String> toRefresh = Sets.newLinkedHashSet();
    Set<String> toAdd = Sets.newLinkedHashSet();
    Set<String> toRemove;

    // 使用synchronized关键字在当前对象上获得互斥锁
    synchronized (this) {
      // Step 1.  针对所有新的nameservices中的每个nameservice，
      // 确认它是一个已存在nameservice中的被更新了的NN集合，还是完全的一个新的nameservice
      // 判断的依据就是对应nameserviceId是否在bpByNameserviceId结合中存在

      // 循环addrMap，放入待添加或者待刷新集合
      for (String nameserviceId : addrMap.keySet()) {
        // 如果bpByNameserviceId结合中存在nameserviceId，加入待刷新集合toRefresh，否则加入到待添加集合toAdd
        if (bpByNameserviceId.containsKey(nameserviceId)) {
          toRefresh.add(nameserviceId);
        } else {
          toAdd.add(nameserviceId);
        }
      }

      // Step 2. 删除所有我们目前拥有但是现在不再需要的，也就是bpByNameserviceId中存在，而配置信息addrMap中没有的

      // 加入到待删除集合toRemove
      toRemove = Sets.newHashSet(Sets.difference(bpByNameserviceId.keySet(), addrMap.keySet()));

      // 验证，待刷新集合toRefresh的大小与待添加集合toAdd的大小必须等于配置信息addrMap中的大小
      assert toRefresh.size() + toAdd.size() ==
        addrMap.size() :
          "toAdd: " + Joiner.on(",").useForNull("<default>").join(toAdd) +
          "  toRemove: " + Joiner.on(",").useForNull("<default>").join(toRemove) +
          "  toRefresh: " + Joiner.on(",").useForNull("<default>").join(toRefresh);

      
      // Step 3. 启动所有新的 nameservices
      if (!toAdd.isEmpty()) {
        LOG.info("Starting BPOfferServices for nameservices: " + Joiner.on(",").useForNull("<default>").join(toAdd));

        // 在这里搞的这个东西，就是 BPOfferService， BPOfferActor,  联邦架构
        // nameservice, 一组namenode, 两个namenode组成的，就是一个active + 一个 standby
        // 一组namenode, 就是一个nameservice
        // 在这里初始化 BlockPoolManager的时候，它其实会根据你的配置文件中的nameservice的配置
        // 对每个nameservice（一组namenode），会去创建一个BPOfferService,

        // 如果你只是用hadoop普通的HA架构，就只有一个nameservice,也就只有一个BPOfferService

        // 针对待添加集合toAdd中的每个nameserviceId，做以下处理：
        for (String nsToAdd : toAdd) {
          // 从addrMap中根据nameserviceId获取对应Socket地址InetSocketAddress，创建集合addrs
          ArrayList<InetSocketAddress> addrs = Lists.newArrayList(addrMap.get(nsToAdd).values());
          // 核心代码: 根据addrs创建BPOfferService
          BPOfferService bpos = createBPOS(addrs); //
          // 将nameserviceId->BPOfferService的对应关系添加到集合bpByNameserviceId中
          bpByNameserviceId.put(nsToAdd, bpos);
          // 将BPOfferService添加到集合offerServices中
          offerServices.add(bpos);
        }
      }

      // startAll()就会去启动对应的BPServiceActor哪些线程, 通过调用它的start()方法启动
      startAll(); //
    }

    // Step 4. 停止所有旧的nameservices。这个是发生在synchronized代码块外面的，是因为它们需要回调另外一个线程的remove()方法
    if (!toRemove.isEmpty()) {
      LOG.info("Stopping BPOfferServices for nameservices: " + Joiner.on(",").useForNull("<default>").join(toRemove));

      // 如果有一些nameservice要删除掉，就要停止它对应的BPServiceActor线程
      // 遍历待删除集合toRemove中的每个nameserviceId
      for (String nsToRemove : toRemove) {
        // 根据nameserviceId从集合bpByNameserviceId中获取BPOfferService
        BPOfferService bpos = bpByNameserviceId.get(nsToRemove);
        // 调用BPOfferService的stop()和join()方法停止服务
        bpos.stop();
        bpos.join();
        // 它们会调用本身的remove()方法
      }
    }
    
    // Step 5. 更新 namenode列表发生变化的nameservices
    if (!toRefresh.isEmpty()) { // 待更新集合toRefresh不为空时
      // 如果某个nameservice的namenode列表变化了，在这里进行处理，
      // 肯定得初始化新的BPServiceActor线程去跟新的namenode进行通信
      LOG.info("Refreshing list of NNs for nameservices: " + Joiner.on(",").useForNull("<default>").join(toRefresh));

      // 遍历待更新集合toRefresh中的每个nameserviceId
      for (String nsToRefresh : toRefresh) {
        // 根据nameserviceId从集合bpByNameserviceId中取出对应的BPOfferService
        BPOfferService bpos = bpByNameserviceId.get(nsToRefresh);
        // 根据BPOfferService从配置信息addrMap中取出NN的Socket地址InetSocketAddress，形成列表addrs
        ArrayList<InetSocketAddress> addrs = Lists.newArrayList(addrMap.get(nsToRefresh).values());
        // 调用BPOfferService的refreshNNList()方法根据addrs刷新NN列表
        bpos.refreshNNList(addrs);
      }
    }
  }

  /**
   * Extracted out for test purposes.
   */
  protected BPOfferService createBPOS(List<InetSocketAddress> nnAddrs) {
    return new BPOfferService(nnAddrs, dn);
  }
}
